/*
 * har_rewrite.c
 *
 *  Created on: Nov 27, 2013
 *      Author: fumin
 */
#include "destor.h"
#include "rewrite_phase.h"
#include "storage/containerstore.h"
#include "jcr.h"
#include "cma.h"
#include "index/index.h"

static pthread_mutex_t mutex;

static GHashTable *container_utilization_monitor;
static GHashTable *inherited_sparse_containers;

extern clear_delta_index();
void statis_func_before_trim();

FILE* f_base_record;

void init_har() {

    pthread_mutex_init(&mutex, NULL);

	/* Monitor the utilizations of containers */
	container_utilization_monitor = g_hash_table_new_full(
			g_int64_hash, g_int64_equal, NULL, free);

    inherited_sparse_containers = g_hash_table_new_full(g_int64_hash,
			g_int64_equal, free, NULL);

    /* record base info (fp and baseid) */
    sds s_base_record = sdsdup(destor.working_directory);
	s_base_record = sdscat(s_base_record, "baseRecord/bv");
	char version[20];
	sprintf(version, "%d", jcr.id);
	s_base_record = sdscat(s_base_record, version);
	s_base_record = sdscat(s_base_record, ".base");

    f_base_record = fopen(s_base_record, "w");
    sdsfree(s_base_record);

    
	/* The first backup doesn't have inherited sparse containers. */
	if (jcr.id > 0) {

        sds fname = sdsdup(destor.working_directory);
		fname = sdscat(fname, "recipes/bv");
		char s[20];
		sprintf(s, "%d", jcr.id - 1);
		fname = sdscat(fname, s);
		fname = sdscat(fname, ".sparse");

		FILE* sparse_file = fopen(fname, "r");

		if (sparse_file) {
			char buf[128];
			while (fgets(buf, 128, sparse_file) != NULL) {
                containerid* p_cid = (containerid*)malloc(sizeof(containerid));
                int32_t size;
				sscanf(buf, "%lld %d", p_cid, &size);

				g_hash_table_insert(inherited_sparse_containers, p_cid,
						NULL);
			}
			fclose(sparse_file);
		}
		sdsfree(fname);
	}

}

void har_monitor_update(containerid id, struct chunk* c) {

    /* monitor both current and base chunks */
    #if 1
	TIMER_DECLARE(1);
	TIMER_BEGIN(1);

	struct containerRecord* record = g_hash_table_lookup(
			container_utilization_monitor, &id);
    
	if (record) 
    {
        if (c->delta)
		{
            record->size += sizeof(c->size) + c->delta->size
                            + sizeof(containerid) + sizeof(fingerprint);
            record->deltanum++;
        }
        else if (c->DepWithDelta && 
                (!CHECK_CHUNK(c, CHUNK_SPARSE) || CHECK_CHUNK(c, CHUNK_REWRITE_DENIED))) {

            assert(c->delta_size != -1);
            record->size += c->delta_size;
            record->deltanum++;
        }
        else {
            record->size += c->size;
		    record->chunknum++;
        }
        
        if (c->delta)
        {
            containerid base_id = c->delta->baseid;

            assert(base_id != TEMPORARY_ID);
            
            struct containerRecord* record_base = g_hash_table_lookup(
			    container_utilization_monitor, &base_id);

            if (record_base)
            {
                record_base->basesize += c->delta->base_size;
                record_base->chunknum++;
                record_base->basechunknum++;
            }
            else
            {
                record_base = (struct containerRecord*) malloc (
					    sizeof(struct containerRecord));
                
		        record_base->basesize = c->delta->base_size;
                record_base->chunknum = 1;
                record_base->deltanum = 0;
                record_base->basechunknum = 1;
                record_base->cid = base_id;
                record_base->size = 0;

                g_hash_table_insert(container_utilization_monitor,
				    &record_base->cid, record_base);
            }
        }
        
	} 
    else 
    {
        record = (struct containerRecord*) malloc(sizeof(struct containerRecord));
        
        if (c->delta)
        {
		    record->size = sizeof(c->size) + c->delta->size
                        + sizeof(containerid) + sizeof(fingerprint);
            record->chunknum = 0;
            record->deltanum = 1;
            record->basechunknum = 0;
            record->cid = id;
            record->basesize = 0;
        }
        else if (c->DepWithDelta && 
                (!CHECK_CHUNK(c, CHUNK_SPARSE) || CHECK_CHUNK(c, CHUNK_REWRITE_DENIED))) {
            
            assert(c->delta_size != -1);
            record->size = c->delta_size;
            record->deltanum = 1;
            record->chunknum = 0;
            record->cid = id;
            record->basesize = 0;
        }
        else {
		    record->size = c->size;
            record->chunknum = 1;
            record->deltanum = 0;
            record->basechunknum = 0;
            record->cid = id;
            record->basesize = 0;
        }
        
        g_hash_table_insert(container_utilization_monitor,
		    &record->cid, record);
        
        if (c->delta)
        {
            containerid base_id = c->delta->baseid;
            
            assert(base_id != TEMPORARY_ID);

            struct containerRecord* record_base = g_hash_table_lookup(
			    container_utilization_monitor, &base_id);

            if (record_base)
            {
                record_base->basesize += c->delta->base_size;
                record_base->chunknum++;
                
                record_base->basechunknum++;
                record_base->basesize += c->delta->base_size;
            }
            else
            {
                record_base = (struct containerRecord*) malloc(
					    sizeof(struct containerRecord));
                
		        record_base->basesize = c->delta->base_size;
                record_base->chunknum = 1;
                record_base->deltanum = 0;
                record_base->basechunknum = 1;
                record_base->cid = base_id;
                record_base->size = 0;
                
                g_hash_table_insert(container_utilization_monitor,
				    &record_base->cid, record_base);
            }
        }
        
    }
	TIMER_END(1, jcr.rewrite_time);
	#endif
}

void DepWithDelta_har_monitor_update(containerid id, struct chunk* c) {
    struct containerRecord* record = g_hash_table_lookup(
			container_utilization_monitor, &id);
    
	if (record) 
    {
        record->size += c->base_size;
        record->chunknum++;
        record->basechunknum++;
    }
    else {
        record = (struct containerRecord*) malloc(
					    sizeof(struct containerRecord));

		record->size = c->base_size;
        record->chunknum = 1;
        record->deltanum = 0;
        record->basechunknum = 1;
        record->cid = id;

        g_hash_table_insert(container_utilization_monitor,
		    &record->cid, record);
    }
}

static gint g_record_cmp(struct containerRecord *a, struct containerRecord* b, gpointer user_data){
	return (a->size + a->basesize) - (b->size + b->basesize);
}

void close_har() {
	sds fname = sdsdup(destor.working_directory);
	fname = sdscat(fname, "recipes/bv");
	char s[20];
	sprintf(s, "%d", jcr.id);
	fname = sdscat(fname, s);
	fname = sdscat(fname, ".sparse");

	FILE* fp = fopen(fname, "w");
    
	if (!fp) {
		fprintf(stderr, "Can not create sparse file");
		perror("The reason is");
		exit(1);
	}

	jcr.total_container_num_referred = g_hash_table_size(container_utilization_monitor);

    GSequence *seq_all = g_sequence_new(NULL);
    
	int64_t total_size = 0;
	int64_t sparse_size = 0;

	/* collect sparse containers */
    GHashTableIter iter;
	gpointer key, value;
	g_hash_table_iter_init(&iter, container_utilization_monitor);
	while (g_hash_table_iter_next(&iter, &key, &value)) {
		struct containerRecord* cr = (struct containerRecord*)value;
		total_size += cr->size;

		if ((1.0*(cr->size + cr->basesize + CONTAINER_HEAD + CONTAINER_META_ENTRY * (cr->chunknum + cr->deltanum))/CONTAINER_SIZE)
				< destor.rewrite_har_utilization_threshold)
	    {
            sparse_size += (cr->size + cr->basesize);
            g_sequence_insert_sorted(seq_all, cr, g_record_cmp, NULL);
		}
	}
    statis_func_before_trim();
	/*
	 * If the size of rewritten data is too large,
	 * we need to trim the sequence to control the rewrite ratio.
	 * We use sparse_size/total_size to estimate the rewrite ratio of next backup.
	 * However, the estimation is inaccurate (generally over-estimating), since:
	 * 	1. the sparse size is not an accurate indicator of utilization for next backup.
	 * 	2. self-references.
	 */
	
    while(destor.rewrite_har_rewrite_limit < 1
			&& sparse_size*1.0/total_size > destor.rewrite_har_rewrite_limit){

		GSequenceIter* iter = g_sequence_iter_prev(g_sequence_get_end_iter(seq_all));
		struct containerRecord* r = g_sequence_get(iter);
		VERBOSE("Trim sparse container %lld", r->cid);
		sparse_size -= (r->size + r->basesize);
		g_sequence_remove(iter);
	}
    
    /* 
     * check if the next version need a full-backup 
     * two conditions to make a full-backup:
     * 1: rewritten chunk size greater than a threshold
     * 2: restore speed factor smaller than a threshold
     * either of these two conditions will trigger a full-backup
     */
    if (destor.enable_full_backup_after_threshold
                        && (destor.full_backup_threshold_of_rewritten > 0)
                        && ((1.0 * jcr.rewritten_chunk_size / (double) jcr.data_size)
                                >= destor.full_backup_threshold_of_rewritten))
            destor.full_backup_next_version = 1;
    else
            destor.full_backup_next_version = 0;

    jcr.predict_restore_speed_factor = jcr.data_size / (1024.0 * 1024 * jcr.containers_to_restore);

    if (destor.enable_full_backup_after_threshold
                        && (destor.restore_speed_factor_to_fullbackup > 0)
                        && (jcr.predict_restore_speed_factor < destor.restore_speed_factor_to_fullbackup))
            destor.full_backup_next_version = 1;
    else
            destor.full_backup_next_version = 0;

    /* write the variable destor.full_backup_next_version */
    sds full_backup_file = sdsdup(destor.working_directory);
	    full_backup_file = sdscat(full_backup_file, "recipes/full.backup");
    FILE* fp_full_backup = fopen(full_backup_file, "w");
    
    if (fp_full_backup) {
        fprintf(fp_full_backup, "%d", destor.full_backup_next_version);
        fclose(fp_full_backup);
    }
    sdsfree(full_backup_file);
    /* end of the written of destor.full_backup_next_version */

    if (destor.enable_full_backup_after_threshold
                && destor.full_backup_next_version 
                && destor.clear_history_info_for_full_backup) {

        clear_delta_index();
        if (!destor.enable_dedup_in_dense_full_backup)
            destor.index_start_point = jcr.container_num_after_backup + 1;
    }
    /* end */

    /* read the index start point */
    sds s_index_start_point_file = sdsdup(destor.working_directory);
	s_index_start_point_file = sdscat(s_index_start_point_file, "index/index.startPoint");
    FILE* fp_index_start_point = fopen(s_index_start_point_file, "w");

    if (fp_index_start_point) {
        fprintf(fp_index_start_point, "%d", destor.index_start_point);
        fclose(fp_index_start_point);
    }
    sdsfree(s_index_start_point_file);
    /* end of check if the next backup need a full-backup */

	GSequenceIter* sparse_iter = g_sequence_get_begin_iter(seq_all);
	while(sparse_iter != g_sequence_get_end_iter(seq_all)){
		struct containerRecord* r = g_sequence_get(sparse_iter);
		fprintf(fp, "%lld %d\n", r->cid, r->size);

        sparse_iter = g_sequence_iter_next(sparse_iter);
	}
	fclose(fp);
    
    pthread_mutex_destroy(&mutex);
    
    g_sequence_free(seq_all);
    
    sdsfree(fname);
    
	/* CMA: update the backup times in manifest */
	update_manifest(container_utilization_monitor);

    g_hash_table_destroy(inherited_sparse_containers);
    g_hash_table_destroy(container_utilization_monitor);

    fclose(f_base_record);
}

void har_check(struct chunk* c) {
	if (!CHECK_CHUNK(c, CHUNK_FILE_START) 
        && !CHECK_CHUNK(c, CHUNK_FILE_END)
	    && (CHECK_CHUNK(c, CHUNK_DUPLICATE) || c->delta)) {
        if (CHECK_CHUNK(c, CHUNK_DUPLICATE)
            && g_hash_table_contains(inherited_sparse_containers, &c->id)) {
			SET_CHUNK(c, CHUNK_SPARSE);
			char code[41];
			hash2code(c->fp, code);
			code[40] = 0;
			VERBOSE("chunk %s in sparse container %ld", code, c->id);
		}
        
        if (c->delta && g_hash_table_contains(inherited_sparse_containers, &c->delta->baseid)) {
            SET_CHUNK(c, CHUNK_SPARSE);
        }
	}
}

void printf_fp_baseid(fingerprint ft, containerid baseid) {
    char code[41];
    hash2code(ft, code);
    code[40] = 0;
    fprintf(f_base_record, "fp: %s, baseid: %lld\n", code, baseid);
}

void statis_func_before_trim() {
    FILE *excel_backup_statis = fopen("excel_before_trim_statis.txt", "a");

    GHashTableIter iter;
	gpointer key, value;
    int64_t total_size = 0;
    int64_t base_referenced_container = 0;
    int64_t normal_chunk_referenced_container = 0;
    int64_t baseOnly_number = 0;
    int64_t fifty_percent_base_container = 0;
    int64_t ten_percent_base_container = 0;
    int64_t sparse_all_container = 0;
    int64_t sparse_chunk_container = 0;
    int64_t avg_size = 0;
    int64_t container_size = 4161536;
    int64_t fragmented_base_chunks = 0;
    int64_t base_only_chunks = 0;

    int64_t base_only_size = 0;

    /* hit ratio of the cache */
    sds s_hit_sparse_base = sdsdup(destor.working_directory);
	s_hit_sparse_base = sdscat(s_hit_sparse_base, "baseRecord/bv");
    char s[20];
	sprintf(s, "%d", jcr.id);
	s_hit_sparse_base = sdscat(s_hit_sparse_base, s);
	s_hit_sparse_base = sdscat(s_hit_sparse_base, ".hit_sparse_base_container");

    FILE* f_hit_sparse_base = fopen(s_hit_sparse_base, "w");
    sdsfree(s_hit_sparse_base);

    int64_t total_containers = g_hash_table_size(container_utilization_monitor);
	g_hash_table_iter_init(&iter, container_utilization_monitor);
	while (g_hash_table_iter_next(&iter, &key, &value)) {
		struct containerRecord* cr = (struct containerRecord*)value;

		if (cr->size == 0)
	    {
            total_size += cr->basesize;
            baseOnly_number++;

            if ((1.0 * cr->basesize / container_size) < 0.5)
                fifty_percent_base_container++;
            
            if ((1.0 * cr->basesize / container_size) < 0.05) {
                ten_percent_base_container++;
                fragmented_base_chunks += cr->basechunknum;
                fprintf(f_hit_sparse_base, "%lld\n", cr->cid);
            }
            base_only_chunks += cr->basechunknum;
            base_only_size += cr->basesize;
		}

        if (cr->basesize > 0)
            base_referenced_container++;

        if (cr->size > 0)
            normal_chunk_referenced_container++;

        if ((1.0*(cr->size + cr->basesize + CONTAINER_HEAD + CONTAINER_META_ENTRY * (cr->chunknum + cr->deltanum))/CONTAINER_SIZE)
				< destor.rewrite_har_utilization_threshold) {
            sparse_all_container++;
		}

        if (cr->basesize == 0 && (1.0*(cr->size + CONTAINER_HEAD + CONTAINER_META_ENTRY * (cr->chunknum + cr->deltanum))/CONTAINER_SIZE)
				< destor.rewrite_har_utilization_threshold) {
		    sparse_chunk_container++;
        }
	}
    fclose(f_hit_sparse_base);
    
    if (baseOnly_number)
        avg_size = total_size/baseOnly_number/1024;
    
    fprintf(excel_backup_statis, "%d\t%lld\t%lld\t%lld\t%lld\t%lld\t%lld\t%lld\t%lld\t%lld\t%.6f\t%.6f\t%.6f\t%.6f\t%.6f\t%.6f\n",
                jcr.id + 1,
                total_containers,
                base_referenced_container,
                baseOnly_number,
                ten_percent_base_container,
                normal_chunk_referenced_container,
                sparse_chunk_container,
                sparse_all_container,
                fragmented_base_chunks,
                base_only_chunks,
                1.0*base_referenced_container/total_containers,
                1.0*normal_chunk_referenced_container/total_containers,
                1.0*fragmented_base_chunks/jcr.number_of_chunks_has_delta,
                1.0*fragmented_base_chunks/ten_percent_base_container,
                1.0*base_only_size/(baseOnly_number*4161536),
                1.0*fifty_percent_base_container/baseOnly_number);
    fclose(excel_backup_statis);
}
